package com.atguigu.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/*
    从本地（windows）向集群(Linux)提交Job
    1.
          //设置在集群运行的相关参数-设置HDFS,NAMENODE的地址
        conf.set("fs.defaultFS", "hdfs://hadoop102:8020");
        //指定MR运行在Yarn上
        conf.set("mapreduce.framework.name","yarn");
        //指定MR可以在远程集群运行
        conf.set("mapreduce.app-submission.cross-platform","true");
        //指定yarn resourcemanager的位置
        conf.set("yarn.resourcemanager.hostname", "hadoop103");

    2.打jar包
    3.注释掉job.setJarByClass(WCDriver3.class);
            添加job.setJar("jar包路径")
    4.
        VM OPtions : -DHADOOP_USER_NAME=atguigu
        Program Arguments : hdfs://hadoop102:8020/input hdfs://hadoop102:8020/output3
 */
public class WCDriver3 {
    public static void main(String[] args) throws Exception {
        //1.创建Job实例
        Configuration conf = new Configuration();
        //设置在集群运行的相关参数-设置HDFS,NAMENODE的地址
        conf.set("fs.defaultFS", "hdfs://hadoop102:8020");
        //指定MR运行在Yarn上
        conf.set("mapreduce.framework.name","yarn");
        //指定MR可以在远程集群运行
        conf.set("mapreduce.app-submission.cross-platform","true");
        //指定yarn resourcemanager的位置
        conf.set("yarn.resourcemanager.hostname", "hadoop103");


        Job job = Job.getInstance(conf);

        //2.给Job赋值
        //设置Jar加载路径--将main方法所在的类传过去（如果是本地运行可以不写）
        //job.setJarByClass(WCDriver3.class);
        job.setJar("D:\\class_video\\0924\\07-Hadoop\\3.代码\\MRDemo\\target\\MRDemo-1.0-SNAPSHOT.jar");

        //设置Mapper和Reducer类
        job.setMapperClass(WCMapper.class);
        job.setReducerClass(WCReducer.class);
        //设置Mapper输出的key,value的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        //设置最终输出的key,value的类型(在这是Reducer输出的key,value的类型)
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        //设置数据的输入输出路径
        FileInputFormat.setInputPaths(job,new Path(args[0]));
        //注意：输出的目录必须不存在
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        //3.运行Job
        /*
        waitForCompletion(boolean verbose)
        verbose : 是否打印用户执行时的信息
        返回值  ：如果为true表示执行成功
         */
        boolean b = job.waitForCompletion(true);
        //退出JVM时的状态：0为正常退出，1为非正常退出
        System.exit(b? 0 : 1);
    }
}
